package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.bean.TaskInfo;
import com.atguigu.upp.bean.TaskTagRule;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.utils.DBUtil;
import com.atguigu.upp.utils.PropertiesUtil;
import com.atguigu.upp.utils.TypeCodeConstant;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/10/13
 *
 *      SparkSql的入口是 SparkSession
 */
public class TaskExecuteApp
{

    //平台会自动传入 taskId, 业务日期
    public static void main(String[] args) throws IOException {

        //假设计算的是1号任务，业务日期是2020-06-14
        Integer taskId = Integer.valueOf(args[0]);
        String doDate = args[1];

        //1.查询要计算的task
        SqlSessionFactory sqlSessionFactory = DBUtil.getSqlSessionFactory("mybatis_mysql.xml");
        SqlSession sqlSession = sqlSessionFactory.openSession();
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSession);

        TaskInfo task = mysqlDBService.getTaskByTaskId(taskId);
        TagInfo tagInfo = mysqlDBService.getTagInfoByTaskId(taskId);
        List<TaskTagRule> tagRules = mysqlDBService.getTaskTagRulesByTaskId(taskId);

       /* System.out.println(task);
        System.out.println(tagInfo);
        System.out.println(tagRules);*/

        //2.为要计算的标签结果建表
        String createTableSql = getCreateTableSql(tagInfo);
        String insertSql = getInsertSql(task, tagInfo, doDate, tagRules);

        //3.向表中导入计算的结果
        SparkSession sparkSession = DBUtil.getSparkSession("TaskExecuteApp");
        sparkSession.sql(createTableSql);
        sparkSession.sql(insertSql);


    }

    /*
        select
    uid,
    x
from (
         SELECT id                                uid,
                `if`(ISNULL(gender), 'U', gender) tagValue
         FROM gmall.dim_user_zip
         WHERE dt = '9999-12-31'
     ) t1


     case tagValue
        when 'M' then '男性'
        when 'F' then '女性'
        when 'U' then '未知'
    end tagValue
     */
    private static String getInsertSql(TaskInfo task,TagInfo tagInfo,String doDate,List<TaskTagRule> tagRules){

        //读取定义的sql
        String sql = task.getTaskSql();
        String replacedSql = sql.replace(';', ' ')
                            .replace("$do_date", doDate);

        //定义语句模板
        String template = " insert overwrite table %s.%s partition (dt='%s') " +
                          " select  uid, %s from ( %s ) t1  ";

        String db = PropertiesUtil.getProperty("updbname");  //库名
        String table = tagInfo.getTagCode().toLowerCase();  //表名

        //判断是不是四级标签，如果是，增加四级标签的映射规则，不是，直接取计算的tagValue
        //判断当前task在 tag_task_rule中有没有封装映射规则，有就是四级标签，否则就不是
        String tagValue = "";
        if (tagRules.size() > 0){

            List<String> result = tagRules
                .stream()
                .map(t -> String.format(" when '%s' then '%s' ", t.getQueryValue(), t.getSubTagValue()))
                .collect(Collectors.toList());

            //把一个集合中的所有字符串拼接为一个字符串
            tagValue =  " case tagValue " +StringUtils.join(result, " ") + "  end tagValue ";

        }else{

            tagValue = "tagValue";
        }

        //格式化
        String insertSql = String.format(template, db, table, doDate, tagValue, replacedSql);

        System.out.println(insertSql);

        return insertSql;


    }

    /*
    模板:
create external table if not exists 库名.标签的tag_code( uid string ,tagValue string|bigint|decimal(16,0))
comment '标签的tag_name|任务的task_info'
partitioned by (dt string)
row format delimited fields terminated by '\t'
location '根路径/表名';
     */
    private static String getCreateTableSql(TagInfo tagInfo){

        String template = " create external table if not exists %s.%s( uid string ,tagValue %s) " +
                          " comment '%s'   " +
                          " partitioned by (dt string)   " +
                          " row format delimited fields terminated by '\t'   " +
                          " location '%s/%s' ";

        //获取相关的参数
        String db = PropertiesUtil.getProperty("updbname");  //库名
        String table = tagInfo.getTagCode().toLowerCase();  //表名
        String tagName = tagInfo.getTagName();              //表注释
        String path = PropertiesUtil.getProperty("hdfsPath"); //表存储在hdfs的根路径


        //根据TagInfo的 tag_value_type 获取标签值的类型:  3：文本，string   1:浮点型 ，decimal(16,0)
        String tagValueType = "";                           //标签列的类型
        switch (tagInfo.getTagValueType()){
            case TypeCodeConstant.TAG_VALUE_TYPE_LONG : tagValueType = "bigint"; break;
            case TypeCodeConstant.TAG_VALUE_TYPE_DECIMAL : tagValueType = "decimal(16,0)"; break;
            case TypeCodeConstant.TAG_VALUE_TYPE_STRING : tagValueType = "string"; break;
            case TypeCodeConstant.TAG_VALUE_TYPE_DATE : tagValueType = "string"; break;
        }

        //填充占位符
        String createTableSql = String.format(template, db, table, tagValueType, tagName, path, table);

        System.out.println(createTableSql);

        return createTableSql;

    }
}
